package com.issac.messagequeue.mqserver.coreclass;

import com.issac.messagequeue.mqserver.VirtualHost;
import com.issac.messagequeue.mqserver.common.Consumer;
import com.issac.messagequeue.mqserver.common.ConsumerEnv;
import com.issac.messagequeue.mqserver.common.MqException;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 实现消费者消费消息的核心逻辑
 */
@Slf4j
public class ConsumerManager {
    private VirtualHost parent;
    // 线程池负责执行具体的回调任务
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;

        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String msgQueueName = tokenQueue.take();
                    // 2. 根据令牌找到queue
                    MSGQueue msgQueue = parent.getMemoryDataCenter().getMsgQueue(msgQueueName);
                    if (msgQueue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现：该队列名不存在！ queueName=" + msgQueueName);
                    }
                    // 3. 从队列消费一个消息
                    synchronized (msgQueue) {
                        consumeMessage(msgQueue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });

        // 把线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }


    public void notifyConsume(String msgQueueName) throws InterruptedException {
        tokenQueue.put(msgQueueName);
    }

    public void addConsumer(String consumerTag, String msgQueueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue msgQueue = parent.getMemoryDataCenter().getMsgQueue(msgQueueName);
        if (msgQueue == null) {
            throw new MqException("[ConsumerManager] 消息队列不存在！ msgQueueName=" + msgQueueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, msgQueueName, autoAck, consumer);
        synchronized (msgQueue) {
            msgQueue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有一些消息了，就要立即消费掉
            int messageCount = parent.getMemoryDataCenter().getMessageCount(msgQueueName);
            for (int i = 0; i < messageCount; i++) {
                consumeMessage(msgQueue);
                log.info("[ConsumerManager] 已从队列消费一条消息！ msgQueue=" + msgQueueName + ", messageCountInMsgQueue" + (messageCount-1));
            }
        }
    }

    private void consumeMessage(MSGQueue msgQueue) {
        // 1. 按照轮询的方式选择出一个消费者
        ConsumerEnv consumerEnv = msgQueue.chooseConsumer();
        if (consumerEnv == null) {
            // 当前还没有消费者
            return;
        }

        // 2. 从队列取出一个消息
        Message message = parent.getMemoryDataCenter().getMessageFromQueue(msgQueue.getName());
        if (message == null) {
            // 当前队列没有消息
            return;
        }
        // 3. 把消息带入到消费者的回调方法，给线程池执行
        workPool.submit(() -> {
            try {
                // 1) 把消息放到待确认集合中
                parent.getMemoryDataCenter().addMessageWaitAck(msgQueue.getName(), message);
                // 2) 执行回调
                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 3） “自动应答” —— 删除消息
                //     “手动应答” —— 先不处理，后续消费者调用 basicAck 处理
                if (consumerEnv.isAutoAck()) {
                    // 删除顺序： 硬盘、待确认集合、内存消息中心
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(message, msgQueue);
                    }
                    parent.getMemoryDataCenter().deleteMessageWaitAck(msgQueue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().deleteMessageById(message.getMessageId());
                    log.error("[ConsumerManager] 消息被消费成功！ msgQueueName=" + msgQueue.getName());
                }
            } catch (Exception e) {
                log.error("[ConsumerManager] 消息被消费失败！ msgQueueName=" + msgQueue.getName());
                e.printStackTrace();
            }
        });
    }
}
